{
 "cells": [
  {
   "metadata": {},
   "cell_type": "markdown",
   "source": [
    "# RxJava 线程调度与背压处理详解\n",
    "\n",
    "下面我将通过一个完整的示例程序，详细展示RxJava的线程调度和背压处理机制，包含丰富的代码示例和日志输出。\n",
    "\n",
    "```java\n",
    "import io.reactivex.rxjava3.core.*;\n",
    "import io.reactivex.rxjava3.disposables.CompositeDisposable;\n",
    "import io.reactivex.rxjava3.functions.*;\n",
    "import io.reactivex.rxjava3.schedulers.Schedulers;\n",
    "import io.reactivex.rxjava3.subscribers.ResourceSubscriber;\n",
    "\n",
    "import java.util.concurrent.TimeUnit;\n",
    "import java.util.concurrent.atomic.AtomicInteger;\n",
    "\n",
    "public class RxJavaThreadingAndBackpressure {\n",
    "\n",
    "    private static final CompositeDisposable disposables = new CompositeDisposable();\n",
    "\n",
    "    public static void main(String[] args) {\n",
    "        System.out.println(\"===== RxJava 线程调度与背压处理示例 =====\");\n",
    "        System.out.println(\"主线程: \" + Thread.currentThread().getName());\n",
    "\n",
    "        // 1. 线程调度基础\n",
    "        basicThreading();\n",
    "\n",
    "        // 2. 多级线程切换\n",
    "        multiLevelThreadSwitching();\n",
    "\n",
    "        // 3. 背压基础示例\n",
    "        basicBackpressure();\n",
    "\n",
    "        // 4. 高级背压策略\n",
    "        advancedBackpressureStrategies();\n",
    "\n",
    "        // 5. 背压与并行处理\n",
    "        backpressureWithParallelProcessing();\n",
    "\n",
    "        // 等待异步操作完成\n",
    "        try {\n",
    "            Thread.sleep(8000);\n",
    "        } catch (InterruptedException e) {\n",
    "            e.printStackTrace();\n",
    "        }\n",
    "        disposables.dispose();\n",
    "    }\n",
    "\n",
    "    // 1. 线程调度基础\n",
    "    private static void basicThreading() {\n",
    "        System.out.println(\"\\n===== 1. 线程调度基础 =====\");\n",
    "\n",
    "        // 创建Observable（默认在当前线程）\n",
    "        Observable<String> observable = Observable.create(emitter -> {\n",
    "            System.out.println(\"发射线程: \" + Thread.currentThread().getName());\n",
    "            emitter.onNext(\"数据1\");\n",
    "            emitter.onNext(\"数据2\");\n",
    "            emitter.onNext(\"数据3\");\n",
    "            emitter.onComplete();\n",
    "        });\n",
    "\n",
    "        // 情况1：不指定线程\n",
    "        observable.subscribe(item ->\n",
    "            System.out.println(\"情况1-接收线程: \" + Thread.currentThread().getName() + \", 数据: \" + item));\n",
    "\n",
    "        // 情况2：只指定subscribeOn\n",
    "        observable\n",
    "            .subscribeOn(Schedulers.io())\n",
    "            .subscribe(item ->\n",
    "                System.out.println(\"情况2-接收线程: \" + Thread.currentThread().getName() + \", 数据: \" + item));\n",
    "\n",
    "        // 情况3：只指定observeOn\n",
    "        observable\n",
    "            .observeOn(Schedulers.computation())\n",
    "            .subscribe(item ->\n",
    "                System.out.println(\"情况3-接收线程: \" + Thread.currentThread().getName() + \", 数据: \" + item));\n",
    "\n",
    "        // 情况4：同时指定subscribeOn和observeOn\n",
    "        observable\n",
    "            .subscribeOn(Schedulers.io())\n",
    "            .observeOn(Schedulers.computation())\n",
    "            .subscribe(item ->\n",
    "                System.out.println(\"情况4-接收线程: \" + Thread.currentThread().getName() + \", 数据: \" + item));\n",
    "    }\n",
    "\n",
    "    // 2. 多级线程切换\n",
    "    private static void multiLevelThreadSwitching() {\n",
    "        System.out.println(\"\\n===== 2. 多级线程切换 =====\");\n",
    "\n",
    "        disposables.add(\n",
    "            Observable.range(1, 5)\n",
    "                .subscribeOn(Schedulers.io()) // 发射在IO线程\n",
    "                .doOnNext(i -> System.out.println(\"发射后: \" + i + \" - 线程: \" + Thread.currentThread().getName()))\n",
    "\n",
    "                .observeOn(Schedulers.computation()) // 切换到计算线程\n",
    "                .map(i -> {\n",
    "                    System.out.println(\"计算线程处理: \" + i + \" - 线程: \" + Thread.currentThread().getName());\n",
    "                    return i * 10;\n",
    "                })\n",
    "\n",
    "                .observeOn(Schedulers.single()) // 再切换到单一线程\n",
    "                .map(i -> {\n",
    "                    System.out.println(\"单一线程处理: \" + i + \" - 线程: \" + Thread.currentThread().getName());\n",
    "                    return \"结果\" + i;\n",
    "                })\n",
    "\n",
    "                .observeOn(Schedulers.io()) // 最后切换到IO线程\n",
    "                .subscribe(\n",
    "                    result -> System.out.println(\"最终结果: \" + result + \" - 线程: \" + Thread.currentThread().getName()),\n",
    "                    Throwable::printStackTrace\n",
    "                )\n",
    "        );\n",
    "    }\n",
    "\n",
    "    // 3. 背压基础示例\n",
    "    private static void basicBackpressure() {\n",
    "        System.out.println(\"\\n===== 3. 背压基础示例 =====\");\n",
    "\n",
    "        // 创建快速发射数据的Flowable\n",
    "        Flowable<Long> fastProducer = Flowable.interval(10, TimeUnit.MILLISECONDS);\n",
    "\n",
    "        // 慢消费者\n",
    "        ResourceSubscriber<Long> slowConsumer = new ResourceSubscriber<Long>() {\n",
    "            @Override\n",
    "            protected void onStart() {\n",
    "                request(1); // 初始请求1个数据\n",
    "            }\n",
    "\n",
    "            @Override\n",
    "            public void onNext(Long item) {\n",
    "                System.out.println(\"处理数据: \" + item + \" - 线程: \" + Thread.currentThread().getName());\n",
    "                try {\n",
    "                    // 模拟慢处理\n",
    "                    Thread.sleep(100);\n",
    "                } catch (InterruptedException e) {\n",
    "                    e.printStackTrace();\n",
    "                }\n",
    "                // 处理完一个数据后请求下一个\n",
    "                request(1);\n",
    "            }\n",
    "\n",
    "            @Override\n",
    "            public void onError(Throwable t) {\n",
    "                System.err.println(\"背压错误: \" + t.getMessage());\n",
    "            }\n",
    "\n",
    "            @Override\n",
    "            public void onComplete() {\n",
    "                System.out.println(\"背压处理完成\");\n",
    "            }\n",
    "        };\n",
    "\n",
    "        // 应用背压策略\n",
    "        disposables.add(\n",
    "            fastProducer\n",
    "                .onBackpressureBuffer( // 设置缓冲区\n",
    "                    50, // 缓冲区大小\n",
    "                    () -> System.out.println(\"缓冲区溢出!\"),\n",
    "                    BackpressureOverflowStrategy.DROP_LATEST\n",
    "                )\n",
    "                .observeOn(Schedulers.computation(), false, 16) // 设置接收缓冲区\n",
    "                .subscribeWith(slowConsumer)\n",
    "        );\n",
    "    }\n",
    "\n",
    "    // 4. 高级背压策略\n",
    "    private static void advancedBackpressureStrategies() {\n",
    "        System.out.println(\"\\n===== 4. 高级背压策略 =====\");\n",
    "\n",
    "        // 策略1: BUFFER - 无限缓冲（可能导致OOM）\n",
    "        Flowable.interval(1, TimeUnit.MILLISECONDS)\n",
    "            .onBackpressureBuffer() // 默认无限缓冲\n",
    "            .observeOn(Schedulers.computation())\n",
    "            .doOnNext(i -> Thread.sleep(10))\n",
    "            .take(10) // 只取10个避免无限运行\n",
    "            .subscribe(i -> System.out.println(\"BUFFER策略: \" + i));\n",
    "\n",
    "        // 策略2: DROP - 丢弃无法处理的数据\n",
    "        Flowable.interval(1, TimeUnit.MILLISECONDS)\n",
    "            .onBackpressureDrop(item ->\n",
    "                System.out.println(\"丢弃数据: \" + item))\n",
    "            .observeOn(Schedulers.computation())\n",
    "            .doOnNext(i -> Thread.sleep(10))\n",
    "            .take(10)\n",
    "            .subscribe(i -> System.out.println(\"DROP策略: \" + i));\n",
    "\n",
    "        // 策略3: LATEST - 只保留最新数据\n",
    "        Flowable.interval(1, TimeUnit.MILLISECONDS)\n",
    "            .onBackpressureLatest()\n",
    "            .observeOn(Schedulers.computation())\n",
    "            .doOnNext(i -> Thread.sleep(10))\n",
    "            .take(10)\n",
    "            .subscribe(i -> System.out.println(\"LATEST策略: \" + i));\n",
    "\n",
    "        // 策略4: ERROR - 缓冲区溢出时抛出异常\n",
    "        Flowable.interval(1, TimeUnit.MILLISECONDS)\n",
    "            .onBackpressureBuffer(16, () -> {}, BackpressureOverflowStrategy.ERROR)\n",
    "            .observeOn(Schedulers.computation())\n",
    "            .doOnNext(i -> Thread.sleep(10))\n",
    "            .take(10)\n",
    "            .subscribe(\n",
    "                i -> System.out.println(\"ERROR策略: \" + i),\n",
    "                err -> System.err.println(\"背压错误: \" + err)\n",
    "            );\n",
    "    }\n",
    "\n",
    "    // 5. 背压与并行处理\n",
    "    private static void backpressureWithParallelProcessing() {\n",
    "        System.out.println(\"\\n===== 5. 背压与并行处理 =====\");\n",
    "\n",
    "        // 模拟大量数据源\n",
    "        Flowable<Integer> dataSource = Flowable.range(1, 1000);\n",
    "\n",
    "        // 使用并行处理 + 背压管理\n",
    "        disposables.add(\n",
    "            dataSource\n",
    "                .parallel(4) // 4个并行流\n",
    "                .runOn(Schedulers.computation()) // 每个流在计算线程上运行\n",
    "                .map(i -> {\n",
    "                    // 模拟耗时处理\n",
    "                    try {\n",
    "                        Thread.sleep(10);\n",
    "                    } catch (InterruptedException e) {\n",
    "                        e.printStackTrace();\n",
    "                    }\n",
    "                    return i * 2;\n",
    "                })\n",
    "                .sequential() // 合并回单个流\n",
    "                .onBackpressureBuffer(100) // 设置缓冲区\n",
    "                .observeOn(Schedulers.single()) // 在单一线程上观察\n",
    "                .subscribe(\n",
    "                    result -> System.out.println(\"并行处理结果: \" + result),\n",
    "                    Throwable::printStackTrace,\n",
    "                    () -> System.out.println(\"并行处理完成\")\n",
    "                )\n",
    "        );\n",
    "    }\n",
    "}\n",
    "```\n",
    "\n",
    "## 线程调度详解\n",
    "\n",
    "### 核心概念\n",
    "1. **subscribeOn()**\n",
    "   - 指定数据发射（上游操作）的执行线程\n",
    "   - 影响范围：从源头到第一个observeOn之前的操作\n",
    "   - 多次调用只有第一次有效\n",
    "\n",
    "2. **observeOn()**\n",
    "   - 指定数据接收（下游操作）的执行线程\n",
    "   - 影响范围：从当前位置到下一个observeOn之前的所有操作\n",
    "   - 可以多次调用，实现多级线程切换\n",
    "\n",
    "### 调度器类型\n",
    "1. `Schedulers.io()`：适用于I/O密集型任务（网络请求、文件操作）\n",
    "2. `Schedulers.computation()`：适用于计算密集型任务（CPU密集型操作）\n",
    "3. `Schedulers.single()`：单一线程，所有任务顺序执行\n",
    "4. `Schedulers.newThread()`：为每个任务创建新线程\n",
    "5. `Schedulers.trampoline()`：在当前线程排队执行\n",
    "\n",
    "### 多级线程切换示例\n",
    "```java\n",
    "Observable.range(1, 5)\n",
    "    .subscribeOn(Schedulers.io()) // 发射在IO线程\n",
    "    .doOnNext(i -> log(\"发射后\", i))\n",
    "\n",
    "    .observeOn(Schedulers.computation()) // 切换到计算线程\n",
    "    .map(i -> {\n",
    "        log(\"计算线程处理\", i);\n",
    "        return i * 10;\n",
    "    })\n",
    "\n",
    "    .observeOn(Schedulers.single()) // 再切换到单一线程\n",
    "    .map(i -> {\n",
    "        log(\"单一线程处理\", i);\n",
    "        return \"结果\" + i;\n",
    "    })\n",
    "\n",
    "    .observeOn(Schedulers.io()) // 最后切换到IO线程\n",
    "    .subscribe(result -> log(\"最终结果\", result));\n",
    "```\n",
    "\n",
    "## 背压处理详解\n",
    "\n",
    "### 背压问题\n",
    "当生产者（被观察者）发射数据的速度快于消费者（观察者）处理数据的速度时，就会产生背压问题。如果不处理，会导致数据积压，最终可能耗尽内存。\n",
    "\n",
    "### 解决方案\n",
    "1. **Flowable**：专为背压设计的Observable变体\n",
    "2. **背压策略**：\n",
    "   - BUFFER：缓存所有数据（可能导致OOM）\n",
    "   - DROP：丢弃无法处理的数据\n",
    "   - LATEST：只保留最新数据\n",
    "   - ERROR：缓冲区溢出时抛出异常\n",
    "\n",
    "### 背压处理示例\n",
    "```java\n",
    "Flowable.interval(1, TimeUnit.MILLISECONDS)\n",
    "    .onBackpressureBuffer( // 设置缓冲区\n",
    "        50, // 缓冲区大小\n",
    "        () -> System.out.println(\"缓冲区溢出!\"), // 溢出回调\n",
    "        BackpressureOverflowStrategy.DROP_LATEST // 策略\n",
    "    )\n",
    "    .observeOn(Schedulers.computation(), false, 16) // 设置接收缓冲区\n",
    "    .subscribe(new ResourceSubscriber<Long>() {\n",
    "        @Override\n",
    "        protected void onStart() {\n",
    "            request(1); // 初始请求1个数据\n",
    "        }\n",
    "\n",
    "        @Override\n",
    "        public void onNext(Long item) {\n",
    "            // 处理数据...\n",
    "            request(1); // 处理完后请求下一个\n",
    "        }\n",
    "    });\n",
    "```\n",
    "\n",
    "### 背压策略比较\n",
    "| 策略        | 特点                                                                 | 适用场景                         |\n",
    "|-------------|----------------------------------------------------------------------|----------------------------------|\n",
    "| BUFFER      | 无限缓存数据，直到消费者处理                                         | 数据量小，不能丢失任何数据的场景 |\n",
    "| DROP        | 丢弃无法处理的数据                                                   | 可以容忍数据丢失的场景           |\n",
    "| LATEST      | 只保留最新数据，覆盖之前未处理的数据                                 | 只需要最新结果的场景             |\n",
    "| ERROR       | 缓冲区满时抛出MissingBackpressureException                           | 需要严格处理背压的场景           |\n",
    "| MISSING     | 无策略，需要下游通过其他方式处理                                     | 高级自定义场景                   |\n",
    "\n",
    "## 程序输出示例\n",
    "\n",
    "```\n",
    "===== RxJava 线程调度与背压处理示例 =====\n",
    "主线程: main\n",
    "\n",
    "===== 1. 线程调度基础 =====\n",
    "发射线程: main\n",
    "情况1-接收线程: main, 数据: 数据1\n",
    "情况1-接收线程: main, 数据: 数据2\n",
    "情况1-接收线程: main, 数据: 数据3\n",
    "发射线程: RxCachedThreadScheduler-1\n",
    "情况2-接收线程: RxCachedThreadScheduler-1, 数据: 数据1\n",
    "情况2-接收线程: RxCachedThreadScheduler-1, 数据: 数据2\n",
    "情况2-接收线程: RxCachedThreadScheduler-1, 数据: 数据3\n",
    "发射线程: main\n",
    "情况3-接收线程: RxComputationThreadPool-1, 数据: 数据1\n",
    "情况3-接收线程: RxComputationThreadPool-1, 数据: 数据2\n",
    "情况3-接收线程: RxComputationThreadPool-1, 数据: 数据3\n",
    "发射线程: RxCachedThreadScheduler-2\n",
    "情况4-接收线程: RxComputationThreadPool-2, 数据: 数据1\n",
    "情况4-接收线程: RxComputationThreadPool-2, 数据: 数据2\n",
    "情况4-接收线程: RxComputationThreadPool-2, 数据: 数据3\n",
    "\n",
    "===== 2. 多级线程切换 =====\n",
    "发射后: 1 - 线程: RxCachedThreadScheduler-3\n",
    "发射后: 2 - 线程: RxCachedThreadScheduler-3\n",
    "发射后: 3 - 线程: RxCachedThreadScheduler-3\n",
    "发射后: 4 - 线程: RxCachedThreadScheduler-3\n",
    "发射后: 5 - 线程: RxCachedThreadScheduler-3\n",
    "计算线程处理: 1 - 线程: RxComputationThreadPool-1\n",
    "计算线程处理: 2 - 线程: RxComputationThreadPool-1\n",
    "计算线程处理: 3 - 线程: RxComputationThreadPool-1\n",
    "计算线程处理: 4 - 线程: RxComputationThreadPool-1\n",
    "计算线程处理: 5 - 线程: RxComputationThreadPool-1\n",
    "单一线程处理: 10 - 线程: RxSingleScheduler-1\n",
    "单一线程处理: 20 - 线程: RxSingleScheduler-1\n",
    "单一线程处理: 30 - 线程: RxSingleScheduler-1\n",
    "单一线程处理: 40 - 线程: RxSingleScheduler-1\n",
    "单一线程处理: 50 - 线程: RxSingleScheduler-1\n",
    "最终结果: 结果10 - 线程: RxCachedThreadScheduler-4\n",
    "最终结果: 结果20 - 线程: RxCachedThreadScheduler-4\n",
    "最终结果: 结果30 - 线程: RxCachedThreadScheduler-4\n",
    "最终结果: 结果40 - 线程: RxCachedThreadScheduler-4\n",
    "最终结果: 结果50 - 线程: RxCachedThreadScheduler-4\n",
    "\n",
    "===== 3. 背压基础示例 =====\n",
    "处理数据: 0 - 线程: RxComputationThreadPool-2\n",
    "处理数据: 1 - 线程: RxComputationThreadPool-2\n",
    "处理数据: 2 - 线程: RxComputationThreadPool-2\n",
    "...\n",
    "缓冲区溢出!\n",
    "...\n",
    "\n",
    "===== 4. 高级背压策略 =====\n",
    "BUFFER策略: 0\n",
    "BUFFER策略: 1\n",
    "...\n",
    "丢弃数据: 10\n",
    "丢弃数据: 11\n",
    "DROP策略: 0\n",
    "DROP策略: 1\n",
    "...\n",
    "LATEST策略: 0\n",
    "LATEST策略: 8\n",
    "背压错误: Buffer is full\n",
    "\n",
    "===== 5. 背压与并行处理 =====\n",
    "并行处理结果: 2\n",
    "并行处理结果: 4\n",
    "并行处理结果: 6\n",
    "...\n",
    "并行处理完成\n",
    "```\n",
    "\n",
    "## 关键点总结\n",
    "\n",
    "1. **线程调度最佳实践**：\n",
    "   - I/O操作使用`Schedulers.io()`\n",
    "   - 计算密集型任务使用`Schedulers.computation()`\n",
    "   - 避免在UI线程执行耗时操作\n",
    "   - 使用`observeOn()`控制UI更新在主线程执行\n",
    "\n",
    "2. **背压处理策略选择**：\n",
    "   - 优先使用`Flowable`处理可能产生背压的场景\n",
    "   - 根据业务需求选择合适的背压策略：\n",
    "     - 重要数据：使用BUFFER（注意设置合理大小）\n",
    "     - 实时数据：使用LATEST\n",
    "     - 可丢弃数据：使用DROP\n",
    "   - 在消费者中合理使用`request(n)`控制消费速率\n",
    "\n",
    "3. **并行处理**：\n",
    "   - 使用`parallel()`和`runOn()`实现并行处理\n",
    "   - 注意并行流合并时的顺序问题\n",
    "   - 结合背压策略处理并行流中的速度不匹配问题\n",
    "\n",
    "通过合理使用线程调度和背压处理，可以构建高效、稳定的响应式应用，有效处理异步数据流中的各种复杂情况。#%% md\n"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Kotlin",
   "language": "kotlin",
   "name": "kotlin"
  },
  "language_info": {
   "name": "kotlin",
   "version": "1.9.23",
   "mimetype": "text/x-kotlin",
   "file_extension": ".kt",
   "pygments_lexer": "kotlin",
   "codemirror_mode": "text/x-kotlin",
   "nbconvert_exporter": ""
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}
